"""
Case Type   : 发布订阅冲突解决
Case Name   : 设置subscription_conflict_resolution=apply_remote, 数据同步过程中订阅端switchover，不影响冲突解决，数据同步
Create At   : 2024/04/12
Owner       : @songjing20
Description :
    1.设置订阅冲突解决方式，开启逻辑复制操主备同步
    2.在两个集群创建表
    3.创建发布订阅
    4.发布、订阅端表中插入数据
    5.订阅端备机switchover升主
    6.订阅端查询表数据
    7.订阅插入数据，发布端更新数据冲突，订阅查询表
    8.订阅恢复主备关系，发布端更新数据，订阅查询
    9.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.成功
    6.200，101
    7.成功，(1001,update)
    8.成功，同步
    9.成功
History     :
"""

import os
import time
import unittest

from yat.test import macro
from yat.test import Node
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 > Primary_SH.get_node_num(), '非1+2以上环境不执行')
class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f"-----{os.path.basename(__file__)} start-----")
        self.pri_userdb_pub = Node(node='PrimaryDbUser')
        self.pri_userdb_sub = Node(node='remote1_PrimaryDbUser')
        self.pri_userdb_sub_std = Node(node='remote1_Standby1DbUser')
        self.constant = Constant()
        self.commsh_pub = CommonSH('PrimaryDbUser')
        self.commsh_sub = CommonSH('remote1_PrimaryDbUser')
        self.commsh_sub_std = CommonSH('remote1_Standby1DbUser')
        self.com_pub = Common()
        self.com_sub = Common('remote1_PrimaryDbUser')
        self.tb_name = 't_pubsub_0349'
        self.pubname = 'pub_pubsub_0349'
        self.subname = 'sub_pubsub_0349'
        self.conflict_param = 'subscription_conflict_resolution'
        self.para_value = self.com_sub.show_param(
            self.conflict_param, macro.DB_ENV_PATH_REMOTE1)
        self.port = str(int(self.pri_userdb_pub.db_port) + 1)
        self.wal_level = self.com_pub.show_param("wal_level")
        self.user_param_pub = f'-U {self.pri_userdb_pub.db_user} ' \
                              f'-W {self.pri_userdb_pub.db_password}'
        self.user_param_sub = f'-U {self.pri_userdb_sub.db_user} ' \
                              f'-W {self.pri_userdb_sub.db_password}'
        text = '-----查询集群状态，修改pg_hba，发布端设置wal_level=logical-----'
        self.log.info(text)
        self.log.info('----- 发布端状态 -----')
        status = self.commsh_pub.get_db_cluster_status('detail')
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, f'执行失败:{text}')
        self.log.info('----- 订阅端状态 -----')
        status = self.commsh_sub.get_db_cluster_status(
            'detail', env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(status)
        self.assertEqual(status.count('Normal'), 4, f'执行失败:{text}')
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host  replication  {self.pri_userdb_sub.db_user}  '
            f'{self.pri_userdb_sub.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, f'发布端配置订阅主机流复制权限失败:{text}')
        guc_res = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host  replication  {self.pri_userdb_sub.db_user}  '
            f'{self.pri_userdb_sub_std.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, f'发布端配置订阅备机流复制权限失败:{text}')
        if 'logical' != self.wal_level:
            res = self.commsh_pub.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG, 'wal_level=logical')
            self.assertTrue(res, f'发布端设置wal_level失败:{text}')
            restart_res = self.commsh_pub.restart_db_cluster()
            self.assertTrue(restart_res, f'重启集群失败:{text}')
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host  replication  {self.pri_userdb_pub.db_user}  '
            f'{self.pri_userdb_pub.db_host}/32 sha256',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(guc_res)
        self.assertTrue(guc_res, f'订阅端配置流复制权限失败:{text}')

    def test_pubsub(self):
        text = '----step1:设置订阅冲突解决方式，开启逻辑复制操主备同步;expect:成功-----'
        self.log.info(text)
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'{self.conflict_param}=apply_remote',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(guc_res, f'订阅端设置参数失败:{text}')
        guc_res = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, 'enable_slot_log = on',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertTrue(guc_res, f'订阅端设置参数失败:{text}')

        text = '-----step2:在两个集群创建表;expect:成功-----'
        self.log.info(text)
        sql = f"""drop table if exists {self.tb_name} cascade;
        create table {self.tb_name}(
          id int primary key, name varchar(20), c3 varchar2(255))
        partition by range(id)(
          partition p1 values less than(30),
          partition p2 values less than(60),
          partition p3 values less than(90),
          partition p4 values less than(maxvalue));"""
        pub_res = self.commsh_pub.execut_db_sql(sql, self.user_param_pub)
        self.log.info(pub_res)
        self.assertEqual(pub_res.count(self.constant.CREATE_TABLE_SUCCESS), 2,
                         f'发布端建表失败:{text}')
        sub_res = self.commsh_sub.execut_db_sql(
            sql, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertEqual(pub_res.count(self.constant.CREATE_TABLE_SUCCESS), 2,
                         f'订阅端建表失败:{text}')

        text = '-----step3:创建发布订阅;expect:成功-----'
        self.log.info(text)
        create_pub = f"drop publication if exists {self.pubname};" \
                     f"create publication {self.pubname} " \
                     f"for table {self.tb_name};"
        pub_res = self.commsh_pub.execut_db_sql(create_pub, self.user_param_pub)
        self.log.info(pub_res)
        self.assertIn(self.constant.create_pub_succ_msg, pub_res,
                      f'执行失败:{text}')
        guc_res = self.commsh_sub.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_keycipher_success, guc_res,
                      f'执行失败:{text}')
        guc_res = self.commsh_sub_std.execute_generate(
            macro.COMMON_PASSWD, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_keycipher_success, guc_res,
                      f'执行失败:{text}')
        create_sub = f"drop subscription if exists {self.subname} cascade;" \
                     f"create subscription {self.subname} connection " \
                     f"'host={self.pri_userdb_pub.db_host} " \
                     f"port={self.port} " \
                     f"user={self.pri_userdb_pub.db_user} " \
                     f"dbname={self.pri_userdb_pub.db_name} " \
                     f"password={self.pri_userdb_pub.ssh_password}' " \
                     f"publication {self.pubname} with(binary=true);"
        sub_res = self.commsh_sub.execut_db_sql(
            create_sub, self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertIn(self.constant.create_sub_succ_msg, sub_res,
                      f'创建订阅失败:{text}')

        text = '-----step4:发布、订阅端表中插入数据;expect:成功-----'
        self.log.info(text)
        sub_res = self.commsh_sub.execut_db_sql(
            f"insert into {self.tb_name} values(generate_series(1, 200, 9), "
            f"'sub_' || generate_series(1, 200, 9), 'sub');",
            self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, sub_res,
                      f'执行失败:{text}')
        time.sleep(1)
        cmd = f"source {macro.DB_ENV_PATH};" \
              f"for i in $(seq 1 200);do " \
              f"gsql -d {self.pri_userdb_pub.db_name} " \
              f"-p {self.pri_userdb_pub.db_port} " \
              f"-c\"insert into {self.tb_name} values($i,'pub_'||$i,'');\";" \
              f"done; gsql -d {self.pri_userdb_pub.db_name} " \
              f"-p {self.pri_userdb_pub.db_port} " \
              f"-c\"update {self.tb_name} set name='update' where id<100;\""
        pub_iu = ComThread(self.pri_userdb_pub.sh, args=(cmd,))
        pub_iu.setDaemon(True)
        pub_iu.start()

        text = '-----step5:订阅端备机switchover升主;expect:成功-----'
        self.log.info(text)
        res = self.commsh_sub_std.execute_gsctl(
            'switchover', self.constant.SWITCH_SUCCESS_MSG,
            env_path=macro.DB_ENV_PATH_REMOTE1,
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1)
        self.log.info(res)
        self.assertTrue(res, f'执行失败:{text}')
        res = self.commsh_sub_std.exec_refresh_conf(macro.DB_ENV_PATH_REMOTE1)
        self.log.info(res)
        self.assertTrue(res, f'执行失败:{text}')
        pub_iu.join(10)
        pub_res = pub_iu.get_result().result()
        self.log.info(pub_res)
        self.assertEqual(pub_res.count(self.constant.INSERT_SUCCESS_MSG), 200,
                         f'执行失败:{text}')
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, pub_res,
                      f'执行失败:{text}')

        text = '-----step6:订阅端查询表数据;expect:200, 101-----'
        self.log.info(text)
        self.commsh_pub.check_sub_consistency(
            self.subname, self.user_param_pub, 20)
        time.sleep(120)
        select_res = self.commsh_sub_std.execut_db_sql(
            f"select count(*) from {self.tb_name};"
            f"select count(*) from {self.tb_name} where name like 'pub%';",
            self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(select_res)
        self.assertIn('200', select_res, f'执行失败:{text}')
        self.assertIn('101', select_res, f'执行失败:{text}')

        text = '-----step7:订阅端插入数据，发布端插入数据冲突，订阅查询表;expect:成功，(1001,update)-----'
        self.log.info(text)
        sub_res = self.commsh_sub_std.execut_db_sql(
            f"insert into {self.tb_name} values(1001, 'sub_1001');",
            self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG, sub_res,
                      f'执行失败:{text}')
        pub_res = self.commsh_pub.execut_db_sql(
            f"update {self.tb_name} set id = 1001 where id =1;",
            self.user_param_pub)
        self.log.info(pub_res)
        self.assertIn(self.constant.UPDATE_SUCCESS_MSG, pub_res,
                      f'执行失败:{text}')
        time.sleep(10)
        sub_res = self.commsh_sub_std.execut_db_sql(
            f"select * from {self.tb_name} where id in(1, 1001);",
            self.user_param_sub, None, macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertIn('1001 | update | \n(1 row)', sub_res, f'执行失败:{text}')

        text = '-----step8:订阅恢复主备关系，发布更新数据，订阅查询;expect:成功，0-----'
        self.log.info(text)
        res = self.commsh_sub.execute_gsctl(
            'switchover', self.constant.SWITCH_SUCCESS_MSG, '', False,
            macro.DB_ENV_PATH_REMOTE1, macro.DB_INSTANCE_PATH_REMOTE1)
        self.log.info(res)
        self.assertTrue(res, f'执行失败:{text}')
        res = self.commsh_sub.exec_refresh_conf(macro.DB_ENV_PATH_REMOTE1)
        self.log.info(res)
        self.assertTrue(res, f'执行失败:{text}')
        pub_res = self.commsh_pub.execut_db_sql(
            f"delete from {self.tb_name};", self.user_param_pub)
        self.log.info(pub_res)
        self.assertIn(self.constant.DELETE_SUCCESS_MSG, pub_res,
                      f'执行失败:{text}')
        time.sleep(10)
        sub_res = self.commsh_sub.execut_db_sql(
            f"select * from {self.tb_name};", self.user_param_sub,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_res)
        self.assertIn('(0 rows)', sub_res, f'执行失败:{text}')

    def tearDown(self):
        text = '-----step9:清理环境;expect:成功-----'
        self.log.info(text)
        res1 = self.commsh_sub.execute_gsctl(
            'switchover', self.constant.SWITCH_SUCCESS_MSG, '', False,
            macro.DB_ENV_PATH_REMOTE1, macro.DB_INSTANCE_PATH_REMOTE1)
        self.log.info(res1)
        res2 = self.commsh_sub.exec_refresh_conf(macro.DB_ENV_PATH_REMOTE1)
        self.log.info(res2)
        drop_sub_res = self.commsh_sub.execut_db_sql(
            f"drop subscription if exists {self.subname};",
            self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_sub_res)
        drop_pub_res = self.commsh_pub.execut_db_sql(
            f"drop publication if exists {self.pubname};", self.user_param_pub)
        self.log.info(drop_pub_res)
        drop_tb = f"drop table if exists {self.tb_name} cascade;"
        drop_tb_sub = self.commsh_sub.execut_db_sql(
            drop_tb, self.user_param_sub, env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info(drop_tb_sub)
        drop_tb_pub = self.commsh_pub.execut_db_sql(
            drop_tb, self.user_param_pub)
        self.log.info(drop_tb_pub)
        wal_level_res = self.commsh_pub.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG,
            f'wal_level={self.wal_level}')
        self.log.info(wal_level_res)
        restart_res = self.commsh_pub.restart_db_cluster()
        pub_guc_res1 = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', 'all', False, False,
            '', f'host    replication  {self.pri_userdb_sub.db_user} '
                f'{self.pri_userdb_sub.db_host}/32 ')
        self.log.info(pub_guc_res1)
        pub_guc_res2 = self.commsh_pub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host  replication  {self.pri_userdb_sub.db_user}  '
            f'{self.pri_userdb_sub_std.db_host}/32 sha256')
        self.log.info(pub_guc_res2)
        sub_guc_res1 = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, macro.DB_INSTANCE_PATH_REMOTE1,
            f'host    replication  {self.pri_userdb_pub.db_user} '
            f'{self.pri_userdb_pub.db_host}/32',
            macro.DB_ENV_PATH_REMOTE1)
        self.log.info(sub_guc_res1)
        sub_guc_res2 = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG,
            f'{self.conflict_param} = {self.para_value}',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        sub_guc_res3 = self.commsh_sub.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, 'enable_slot_log = off',
            dn_path=macro.DB_INSTANCE_PATH_REMOTE1,
            env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub_res,
                      f'删除发布失败:{text}')
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub_res,
                      f'删除订阅失败:{text}')
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_tb_pub,
                      f'发布端删表失败:{text}')
        self.assertIn(self.constant.TABLE_DROP_SUCCESS, drop_tb_sub,
                      f'订阅端删表失败:{text}')
        self.assertTrue(wal_level_res, f'执行失败:{text}')
        self.assertTrue(restart_res, f'重启发布端集群失败:{text}')
        self.assertTrue(pub_guc_res1 and pub_guc_res2, f'执行失败:{text}')
        self.assertTrue(sub_guc_res1 and sub_guc_res2 and sub_guc_res3,
                        f'执行失败:{text}')
        self.assertTrue(res1 and res2, f'执行失败:{text}')
        self.log.info(f"-----{os.path.basename(__file__)} end-----")
